package org.fight.landowner.remoting.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.fight.landowner.common.DelayExecuteTaskQueue;

import org.fight.landowner.common.util.OSUtil;
import org.fight.landowner.remoting.RemotingServiceStatus;
import org.fight.landowner.remoting.exception.RemotingServiceStatusException;
import org.fight.landowner.remoting.protocol.RemotingCommand;

/**
 * @Author wangxuehao02
 * @Date 2024/2/1 8:34 PM
 * @Description
 **/
public class NettyRemotingServer extends NettyAbstractRemotingService {

    private int port;

    private ServerBootstrap serverBootstrap;

    private Channel channel;

    private EventLoopGroup bossEventLoopGroup;

    private EventLoopGroup workEventLoopGroup;

    private static NettyServerHandler nettyServerHandler;

    public NettyRemotingServer(int port) {
        this.port = port;

        boolean isLinux = OSUtil.isLinux();

        this.bossEventLoopGroup = isLinux ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        this.workEventLoopGroup = isLinux ? new EpollEventLoopGroup() : new NioEventLoopGroup();

        this.serverBootstrap = new ServerBootstrap();

        NettyRemotingServer.this.nettyServerHandler = new NettyServerHandler();

        this.serverBootstrap
                .channel(isLinux ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .group(bossEventLoopGroup, workEventLoopGroup)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) {
                        socketChannel
                                .pipeline()
                                .addLast(new NettyDecoder())
                                .addLast(NettyEncoder.getInstance())
                                .addLast(NettyRemotingServer.this.nettyServerHandler); // todo 一个可复用
                    }
                });

        this.responseDelayQueue = new DelayExecuteTaskQueue("NettyRemotingServerResponseDelayQueue", new DelayExecuteTaskQueue.DelayExecuteTaskActuator<ResponseCleanDelayTask>() {
            @Override
            public void execute(ResponseCleanDelayTask delayTask) {
                handlerTimeoutRequest(delayTask.getRequestId());
            }
        });
    }

    public void start() throws InterruptedException {
        if (!status.compareAndSet(RemotingServiceStatus.INIT.getCode(), RemotingServiceStatus.RUNNING.getCode())) {
            throw new RemotingServiceStatusException("NettyRemotingServer status is not init");
        }

        ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
        this.channel = channelFuture.channel();

        this.responseDelayQueue.start();
    }

    /**
     * 需不需要等待被 close
     */
    public void shutdown() throws InterruptedException {
        if (status.compareAndSet(RemotingServiceStatus.INIT.getCode(), RemotingServiceStatus.STOP.getCode())) {
            return;
        }

        if (status.compareAndSet(RemotingServiceStatus.RUNNING.getCode(), RemotingServiceStatus.STOP.getCode())) {
            this.channel.close().sync();
            this.bossEventLoopGroup.shutdownGracefully();
            this.workEventLoopGroup.shutdownGracefully();
            this.responseDelayQueue.stop();
        }
    }

    public int getPort() {
        return port;
    }

    @ChannelHandler.Sharable
    private class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) throws Exception {
            processRemotingCommand(channelHandlerContext.channel(), remotingCommand);
        }
    }
}
